package com.timing.finder.framework;

import com.example.autojob.logging.model.producer.AutoJobLogHelper;
import com.timing.finder.util.ProtoStuffUtil;
import com.timing.finder.util.RedisUtil;
import com.timing.finder.util.StringUtils;
import io.lettuce.core.KeyScanCursor;
import io.lettuce.core.ScanArgs;
import io.lettuce.core.cluster.api.async.RedisAdvancedClusterAsyncCommands;
import io.lettuce.core.cluster.api.sync.RedisAdvancedClusterCommands;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.RedisPipelineException;
import org.springframework.data.redis.connection.RedisStringCommands;
import org.springframework.data.redis.connection.lettuce.LettuceClusterConnection;
import org.springframework.data.redis.core.ConvertingCursor;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.data.redis.core.types.Expiration;
import org.springframework.data.redis.serializer.RedisSerializer;

import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * @author JingGe(* ^ ▽ ^ *)
 * @date 2023-03-06 14:12
 * @email 1158055613@qq.com
 */
@Slf4j
public class RedisRepository implements CacheRepository {
    private final CacheStoreSerializer<byte[]> storeSerializer;
    private final CacheStoreDeserializer<byte[]> storeDeserializer;
    private final TimingFinderConfig config;


    public RedisRepository(CacheStoreSerializer<byte[]> storeSerializer, CacheStoreDeserializer<byte[]> storeDeserializer, TimingFinderConfig config) {
        this.storeSerializer = storeSerializer;
        this.storeDeserializer = storeDeserializer;
        this.config = config;
    }

    @Override
    public String formatKey(String key) {
        return String.format("%s:%s", config.getApplicationName(), key);
    }

    @Override
    public <T> boolean set(String key, T value) {
        try {
            RedisUtil
                    .getRedisTemplate()
                    .execute(connection -> connection.set(defaultRawKey(key), ProtoStuffUtil.serialize(value)), true);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    @Override
    public <T> boolean set(String key, T value, long expiringTime, TimeUnit unit) {
        try {
            //要用自己的序列化器序列化Value一定要使用execute并且exposeConnection要设置成true，否则返回的是connection的代理
            RedisUtil
                    .getRedisTemplate()
                    .execute(connection -> connection.set(defaultRawKey(key), ProtoStuffUtil.serialize(value), Expiration.from(expiringTime, unit), RedisStringCommands.SetOption.UPSERT), true);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    @Override
    public <T> T get(String key, Class<T> type) {
        return ProtoStuffUtil.deserialize(RedisUtil
                .getRedisTemplate()
                .execute(connection -> connection.get(defaultRawKey(key)), true), type);
    }

    @Override
    public long alloc(String key, long size) {
        if (!isExist(key)) {
            return 0;
        }
        byte[][] nullData = new byte[(int) size][];
        for (int i = 0; i < size; i++) {
            nullData[i] = new byte[]{};
        }
        List<Object> result = RedisUtil
                .getRedisTemplate()
                .executePipelined((RedisCallback<Object>) connection -> {
                    connection.openPipeline();
                    try {
                        connection
                                .listCommands()
                                .rPush(defaultRawKey(key), nullData);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                    return null;
                });
        log.debug("扩容完成，{}目前长度为{}", key, result.get(0));
        return (long) result.get(0);
    }

    @Override
    public boolean isExist(String key) {
        Boolean flag = RedisUtil
                .getRedisTemplate()
                .hasKey(key);
        return flag != null && flag;
    }

    @Override
    public boolean createKeyListIfAbsent(String key, long startOffset, long size, long expiringTime, TimeUnit unit) {
        if (isExist(key)) {
            log.info("key{}已存在长度为{}的桶", key, size(key));
            return true;
        }
        Long length = RedisUtil
                .getRedisTemplate()
                .opsForList()
                .rightPush(key, storeSerializer.serialize(new StartOffset(startOffset)));
        RedisUtil
                .getRedisTemplate()
                .expire(key, expiringTime, unit);
        //初始化整个List
        alloc(key, size);
        return length != null;
    }

    @Override
    public boolean createKeyListsIfAbsent(Map<String, Long> keyMap, long startOffset, long expiringTime, TimeUnit unit) {
        try {
            RedisUtil
                    .getRedisTemplate()
                    .executePipelined((RedisCallback<Object>) connection -> {
                        connection.openPipeline();
                        for (Map.Entry<String, Long> entry : keyMap.entrySet()) {
                            byte[][] nullData = new byte[entry
                                    .getValue()
                                    .intValue() + 1][];
                            for (int i = 1; i <= entry.getValue(); i++) {
                                nullData[i] = new byte[]{};
                            }
                            nullData[0] = storeSerializer.serialize(new StartOffset(startOffset));
                            connection
                                    .listCommands()
                                    .rPush(defaultRawKey(entry.getKey()), nullData);
                            connection.expire(defaultRawKey(entry.getKey()), unit.toSeconds(expiringTime));
                            getBloomFilter().put(entry.getKey());
                        }
                        return null;
                    });
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
        return true;
    }

    @Override
    public <R extends TimeSeriesData> int loadBatch(Map<String, Long> keyMap, List<ValueEntry<R>> valueEntries, long startOffset, long expiringTime, TimeUnit unit) {
        AutoJobLogHelper logHelper = AutoJobLogHelper.getInstance();
        logHelper.setSlf4jProxy(log);
        AtomicInteger count = new AtomicInteger(0);
        long start = System.currentTimeMillis();

        StorableBloomFilter bloomFilter = getBloomFilter();
        Map<String, Boolean> existKeys = new HashMap<>();
        int existCount = 0;
        for (String key : keyMap.keySet()) {
            if (bloomFilter.mightContains(key)) {
                existCount++;
                existKeys.put(key, true);
            } else {
                existKeys.put(key, false);
            }
        }
        logHelper.info("线程{}判断key存在性用时{}ms，筛选出{}/{}个已存在桶", Thread
                .currentThread()
                .getName(), System.currentTimeMillis() - start, existCount, keyMap.size());
        start = System.currentTimeMillis();

        try {
            RedisUtil
                    .getRedisTemplate()
                    .executePipelined((RedisCallback<Object>) connection -> {
                        connection.openPipeline();
                        for (Map.Entry<String, Long> entry : keyMap.entrySet()) {
                            if (existKeys.containsKey(entry.getKey()) && existKeys.get(entry.getKey())) {
                                //log.info("桶{}已存在，将不会被创建", entry.getKey());
                                continue;
                            }
                            byte[][] nullData = new byte[entry
                                    .getValue()
                                    .intValue() + 1][];
                            for (int i = 1; i <= entry.getValue(); i++) {
                                nullData[i] = new byte[]{};
                            }
                            nullData[0] = storeSerializer.serialize(new StartOffset(startOffset));
                            connection
                                    .listCommands()
                                    .rPush(defaultRawKey(entry.getKey()), nullData);
                            connection.expire(defaultRawKey(entry.getKey()), unit.toSeconds(expiringTime));
                            bloomFilter.put(entry.getKey());
                        }
                        return null;
                    });
        } catch (Exception e) {
            throw new BucketCreateException("桶创建异常" + e.getMessage());
        }
        logHelper.info("线程{}创建桶用时{}ms", Thread
                .currentThread()
                .getName(), System.currentTimeMillis() - start);
        start = System.currentTimeMillis();

        try {
            RedisUtil
                    .getRedisTemplate()
                    .executePipelined((RedisCallback<Object>) connection -> {
                        connection.openPipeline();
                        for (ValueEntry<R> entry1 : valueEntries) {
                            if (entry1.getIndex() >= 0 && entry1.getIndex() <= config.initialSize + 1) {
                                connection
                                        .listCommands()
                                        .lSet(defaultRawKey(entry1.getKey()), entry1.getIndex(), storeSerializer.serialize(entry1
                                                .getValue()
                                                .storeValue()));
                            } else {
                                logHelper.warn("{}偏移异常：{}", entry1.getKey(), entry1.getIndex());
                            }
                            count.incrementAndGet();
                        }
                        return null;
                    });
        } catch (RedisPipelineException e) {
            throw new DataSaveException("数据保存异常" + e.getMessage());
        }
        logHelper.info("线程{}保存数据用时{}ms", Thread
                .currentThread()
                .getName(), System.currentTimeMillis() - start);
        return count.get();
    }

    @Override
    //@SuppressWarnings("unchecked")
    public Map<String, Boolean> scanKeysExist(List<String> keys) {
        AutoJobLogHelper logHelper = AutoJobLogHelper.getInstance();
        logHelper.setSlf4jProxy(log);
        Map<String, Boolean> exist = new HashMap<>();
        keys.forEach(key -> exist.put(key, false));
        scanPatternKeys(null).forEach(key -> {
            if (exist.containsKey(key)) {
                exist.put(key, true);
            }
        });
        return exist;
        //int count = 0;
        //try {
        //    ScanOptions options = ScanOptions
        //            .scanOptions()
        //            .count(Integer.MAX_VALUE)
        //            .build();
        //    RedisSerializer<String> redisSerializer = (RedisSerializer<String>) RedisUtil
        //            .getRedisTemplate()
        //            .getKeySerializer();
        //    Cursor<String> cursor = RedisUtil
        //            .getRedisTemplate()
        //            .executeWithStickyConnection(redisConnection -> new ConvertingCursor<>(redisConnection.scan(options), redisSerializer::deserialize));
        //    if (cursor != null) {
        //        while (cursor.hasNext()) {
        //            String key = "";
        //            count++;
        //            try {
        //                key = cursor.next();
        //            } catch (Exception ignored) {
        //                continue;
        //            }
        //            if (exist.containsKey(key)) {
        //                exist.put(key, true);
        //            }
        //        }
        //        cursor.close();
        //    }
        //} catch (Exception e) {
        //    throw new ScanBucketException("扫描桶时发生异常" + e.getMessage());
        //} finally {
        //    logHelper.info("本次累计扫描key{}个", count);
        //}
        //return exist;
    }

    @Override
    @SuppressWarnings("unchecked")
    public Set<String> scanPatternKeys(String pattern) {
        AutoJobLogHelper logHelper = AutoJobLogHelper.getInstance();
        logHelper.setSlf4jProxy(log);
        Set<String> keys = new HashSet<>();
        RedisSerializer<String> redisSerializer = (RedisSerializer<String>) RedisUtil
                .getRedisTemplate()
                .getKeySerializer();
        if (config.isCacheClustered) {
            try {
                RedisConnectionFactory connectionFactory = RedisUtil
                        .getRedisTemplate()
                        .getConnectionFactory();
                if (connectionFactory == null) {
                    return keys;
                }
                LettuceClusterConnection con = (LettuceClusterConnection) connectionFactory.getClusterConnection();
                RedisAdvancedClusterAsyncCommands<byte[], byte[]> nativeConnection = (RedisAdvancedClusterAsyncCommands<byte[], byte[]>) con.getNativeConnection();
                RedisAdvancedClusterCommands<byte[], byte[]> sync = nativeConnection
                        .getStatefulConnection()
                        .sync();
                KeyScanCursor<byte[]> scanCursor = null;
                ScanArgs scanArgs = new ScanArgs();
                scanArgs.limit(100000);
                if (!StringUtils.isEmpty(pattern)) {
                    scanArgs.match(pattern);
                }
                do {
                    if (scanCursor == null) {
                        scanCursor = sync.scan(scanArgs);
                    } else {
                        scanCursor = sync.scan(scanCursor, scanArgs);
                    }
                    keys.addAll(scanCursor
                            .getKeys()
                            .stream()
                            .map(key -> {
                                try {
                                    return redisSerializer.deserialize(key);
                                } catch (Exception ignored) {
                                }
                                return null;
                            })
                            .filter(Objects::nonNull)
                            .collect(Collectors.toList()));
                } while (!scanCursor.isFinished());
            } catch (Exception e) {
                throw new ScanBucketException("扫描桶时发生异常：" + e.getMessage());
            }
        } else {
            try {
                ScanOptions options = ScanOptions
                        .scanOptions()
                        .match(pattern)
                        .count(100000)
                        .build();
                Cursor<String> cursor = RedisUtil
                        .getRedisTemplate()
                        .executeWithStickyConnection(redisConnection -> new ConvertingCursor<>(redisConnection.scan(options), redisSerializer::deserialize));
                if (cursor != null) {
                    while (cursor.hasNext()) {
                        String key = "";
                        try {
                            key = cursor.next();
                        } catch (Exception ignored) {
                            continue;
                        }
                        keys.add(key);
                    }
                    cursor.close();
                }
            } catch (Exception e) {
                throw new ScanBucketException("扫描桶时发生异常" + e.getMessage());
            }
        }
        logHelper.info("本次遍历key{}个", keys.size());
        return keys;
    }

    @Override
    public long deletePattern(String pattern) {
        Set<String> waitingDelete = scanPatternKeys(pattern);
        Long delCount = RedisUtil
                .getRedisTemplate()
                .delete(waitingDelete);
        waitingDelete.clear();
        System.gc();
        return delCount == null ? 0 : delCount;
    }

    @Override
    public long startOffset(String key) {
        if (size(key) <= 0) {
            return -1;
        }
        byte[] res = RedisUtil
                .getRedisTemplate()
                .execute(connection -> connection
                        .listCommands()
                        .lIndex(defaultRawKey(key), 1L), true);
        if (res == null || res.length == 0) {
            return -1;
        }
        StartOffset startOffset = storeDeserializer.deserialize(res, StartOffset.class);
        return startOffset != null && startOffset.getStartTimestamp() != 0 ? startOffset.getStartTimestamp() : -1;
    }

    @Override
    public <R extends TimeSeriesData> boolean indexSet(String key, long index, R value) {
        try {
            if (index < 0) {
                return false;
            }
            RedisUtil
                    .getRedisTemplate()
                    .opsForList()
                    .set(key, index, storeSerializer.serialize(value.storeValue()));
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
        return true;
    }

    @Override
    public <R extends TimeSeriesData> int indexSets(String key, Map<Long, R> values) {
        AtomicInteger count = new AtomicInteger(0);
        RedisUtil
                .getRedisTemplate()
                .executePipelined((RedisCallback<Object>) connection -> {
                    connection.openPipeline();
                    for (Map.Entry<Long, R> entry : values.entrySet()) {
                        if (entry.getKey() < 0) {
                            continue;
                        }
                        connection
                                .listCommands()
                                .lSet(defaultRawKey(key), entry.getKey(), storeSerializer.serialize(entry
                                        .getValue()
                                        .storeValue()));
                        count.incrementAndGet();
                    }
                    return null;
                });
        return count.get();
    }

    @Override
    public <T extends CacheData> T index(String key, long index, Class<T> type) {
        long size = size(key);
        if (index > size || index < 0) {
            return null;
        }
        byte[] res = RedisUtil
                .getRedisTemplate()
                .execute(connection -> connection
                        .listCommands()
                        .lIndex(defaultRawKey(key), index), true);
        if (res == null || res.length == 0) {
            return null;
        }
        return (T) storeDeserializer.deserialize(res, type);
    }

    @Override
    public <T extends CacheData> List<T> range(String key, long start, long end, Class<T> type) {
        long size = size(key);
        if (size < 0) {
            return Collections.emptyList();
        }
        List<byte[]> res = RedisUtil
                .getRedisTemplate()
                .execute(connection -> connection
                        .listCommands()
                        .lRange(defaultRawKey(key), start, end), true);
        if (res == null || res.size() == 0) {
            return Collections.emptyList();
        }
        List<T> result = new ArrayList<>();
        for (byte[] data : res) {
            if (data == null || data.length == 0) {
                continue;
            }
            result.add(storeDeserializer.deserialize(data, type));
        }
        return result;
    }


    @Override
    public long size(String key) {
        Long size = RedisUtil
                .getRedisTemplate()
                .opsForList()
                .size(key);
        return size == null ? 0 : size;
    }


    @SuppressWarnings("unchecked")
    private byte[] defaultRawKey(String key) {
        return Objects.requireNonNull(((RedisSerializer<String>) RedisUtil
                .getRedisTemplate()
                .getKeySerializer()).serialize(key));
        //return ProtoStuffUtil.serialize(key);
    }

    private StorableBloomFilter getBloomFilter() {
        return BloomFilterContainer
                .getInstance()
                .getBloomFilter(config.applicationName);
    }

}
